package yuekao9.ads;

import yuekao9.util.KafkaUtil;
import yuekao9.entity.OrderInfo;
import com.alibaba.fastjson.JSON;
import yuekao9.entity.OrderTraceLog;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;

public class ComputeData {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<String> streamSource = env.addSource(KafkaUtil.kafkaSource("tms_ods"));
//        streamSource.print();
        //1.按小时统计物流订单在“待支付”、“待分拣”“待签“待取件"收”的状态下的占比(5分)
        OutputTag<String> order_trace_log = new OutputTag<String>("order_trace_log") {
        };
        SingleOutputStreamOperator<String> orderinfodata = streamSource.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String s, ProcessFunction<String, String>.Context context, Collector<String> collector) throws Exception {
                String table = JSON.parseObject(s).getString("table");
                if ("order_info".equals(table)) {
                    collector.collect(s);
                }
                if("order_trace_log".equals(table)){
                    context.output(order_trace_log,s);
                }
            }
        });
//        orderinfodata.print();

        SingleOutputStreamOperator<OrderInfo> orderinfoDS = orderinfodata.map(new MapFunction<String, OrderInfo>() {
            @Override
            public OrderInfo map(String s) throws Exception {
                String data = JSON.parseObject(s).getString("data");
                return JSON.parseObject(data, OrderInfo.class);
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy
                .<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(20))
                .withTimestampAssigner((event, timestamp) -> {
                    try {
                        return new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'").parse(event.getCreate_time()).getTime();
                    } catch (ParseException e) {
                        throw new RuntimeException(e);
                    }
                }));

//        orderinfoDS.print();

        SingleOutputStreamOperator<OrderTraceLog> OrderTraceLogDS = orderinfodata.getSideOutput(order_trace_log).map(new MapFunction<String, OrderTraceLog>() {
            @Override
            public OrderTraceLog map(String s) throws Exception {
                String data = JSON.parseObject(s).getString("data");
                return JSON.parseObject(data, OrderTraceLog.class);
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy
                .<OrderTraceLog>forBoundedOutOfOrderness(Duration.ofSeconds(20))
                .withTimestampAssigner((event, timestamp) -> {
                    try {
                        return new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'").parse(event.getCreate_time()).getTime();
                    } catch (ParseException e) {
                        throw new RuntimeException(e);
                    }
                }));

        SingleOutputStreamOperator<Tuple3<String,Long, Long>> process = orderinfoDS
                .keyBy(x -> x.getId())
                .intervalJoin(OrderTraceLogDS.keyBy(x -> x.getOrder_id()))
                .between(Time.days(-30), Time.days(30))
                .process(new ProcessJoinFunction<OrderInfo, OrderTraceLog, Tuple3<String,Long, Long>>() {
                    @Override
                    public void processElement(OrderInfo orderInfo, OrderTraceLog orderTraceLog,
                                               ProcessJoinFunction<OrderInfo, OrderTraceLog, Tuple3<String,Long, Long>>.Context context,
                                               Collector<Tuple3<String,Long, Long>> collector) throws Exception {
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
                        Long time1 = 0L;
                        Long time2 = 0L;
                        if(orderInfo.getCreate_time()!=null&&orderTraceLog.getCreate_time()!=null){
                            if ("60010".equals(orderInfo.getStatus())) {
                                time1 = sdf.parse(orderInfo.getCreate_time()).getTime();
                            }
                            if ("已取件".equals(orderTraceLog.getTrace_desc())) {
                                time2 = sdf.parse(orderTraceLog.getCreate_time()).getTime();
                            }
                        }
                        collector.collect(new Tuple3<>(orderInfo.getId(),time1, time2));
                    }
                });

        process.print();

        SingleOutputStreamOperator<Tuple2<OrderInfo, OrderTraceLog>> process1 = orderinfoDS
                .keyBy(x -> x.getId())
                .intervalJoin(OrderTraceLogDS.keyBy(x -> x.getOrder_id()))
                .between(Time.days(-30), Time.days(30))
                .process(new ProcessJoinFunction<OrderInfo, OrderTraceLog, Tuple2<OrderInfo, OrderTraceLog>>() {
                    @Override
                    public void processElement(OrderInfo orderInfo, OrderTraceLog orderTraceLog, ProcessJoinFunction<OrderInfo, OrderTraceLog, Tuple2<OrderInfo, OrderTraceLog>>.Context context, Collector<Tuple2<OrderInfo, OrderTraceLog>> collector) throws Exception {
                        collector.collect(new Tuple2<>(orderInfo, orderTraceLog));
                    }
                });
//        process1.print();
        //2.按城市统计物流订单的取件耗时排名(取件时长=上门取件时间-用户下单时间)(5分)
        //3、结合下单、取件、分拣、转运字段，自主设计物流健康分，并计算各转运站的物流健康分(10分)
        //4、自主设计规则计算物流单的ETA(预计到达时间)(6分)

        env.execute();
    }
}
